package com.lzq.mq.common.remote;

import com.alibaba.fastjson2.JSON;
import com.lzq.mq.common.cache.BrokerServerSyncFutureManage;
import com.lzq.mq.common.coder.TcpMsg;
import com.lzq.mq.common.dto.ClientSendMessageToBrokerRespDTO;
import com.lzq.mq.common.dto.ConsumeMsgAckRespDTO;
import com.lzq.mq.common.dto.ConsumeMsgRespDTO;
import com.lzq.mq.common.enums.BrokerResponseCode;
import com.lzq.mq.common.enums.ClientSendMessageToBrokerRespStatus;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import lombok.extern.slf4j.Slf4j;

/**
 * broker客户带端响应处理
 */
@Slf4j
@ChannelHandler.Sharable
public class BrokerRemoteRespHandler extends SimpleChannelInboundHandler<TcpMsg> {

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TcpMsg msg) throws Exception {

        if (msg.getCode() == BrokerResponseCode.SEND_MSG_RESP.getCode()) {
            ClientSendMessageToBrokerRespDTO respDTO = JSON.parseObject(msg.getBody(), ClientSendMessageToBrokerRespDTO.class);
            if (respDTO.getMsgId() == null) {
                // 为空的不需要处理啊！
                return;
            }
            SyncFuture syncFuture = BrokerServerSyncFutureManage.get(respDTO.getMsgId());
            if (syncFuture != null) {
                syncFuture.setResponse(msg);
            }
        } else if (BrokerResponseCode.CONSUME_MSG_RESP.getCode() == msg.getCode()) {
            ConsumeMsgRespDTO consumeMsgRespDTO = JSON.parseObject(msg.getBody(), ConsumeMsgRespDTO.class);
            SyncFuture syncFuture = BrokerServerSyncFutureManage.get(consumeMsgRespDTO.getMsgId());
            //log.info("BrokerRemoteRespHandler收到broker的消费消息响应:{}", consumeMsgRespDTO.getMsgId());
            if (syncFuture != null) {
                syncFuture.setResponse(msg);
            }
        } else if (BrokerResponseCode.BROKER_UPDATE_CONSUME_OFFSET_RESP.getCode() == msg.getCode()) {
            ConsumeMsgAckRespDTO consumeMsgAckRespDTO = JSON.parseObject(msg.getBody(), ConsumeMsgAckRespDTO.class);
            SyncFuture syncFuture = BrokerServerSyncFutureManage.get(consumeMsgAckRespDTO.getMsgId());
            if (syncFuture != null) {
                syncFuture.setResponse(msg);
            }
        } else if (BrokerResponseCode.CREATED_TOPIC_SUCCESS.getCode() == msg.getCode()) {
            ConsumeMsgRespDTO dto = JSON.parseObject(msg.getBody(), ConsumeMsgRespDTO.class);
            SyncFuture syncFuture = BrokerServerSyncFutureManage.get(dto.getMsgId());
            if (syncFuture != null) {
                syncFuture.setResponse(msg);
            }
        } else if (BrokerResponseCode.START_SYNC_SUCCESS.getCode() == msg.getCode()) {
            ConsumeMsgRespDTO dto = JSON.parseObject(msg.getBody(), ConsumeMsgRespDTO.class);
            SyncFuture syncFuture = BrokerServerSyncFutureManage.get(dto.getMsgId());
            if (syncFuture != null) {
                syncFuture.setResponse(msg);
            }
        }

    }    /**
     * 断开连接处理
     */
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        super.channelInactive(ctx);
        // 如果依赖任务剔除节点, 会有三个心跳周期的延迟
        // 想要做到链接断开, 立即剔除, 可以通过这个回调函数发送UnRegister时间
        log.info("断开与Broker的连接:{}", ctx.channel().remoteAddress());
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.error("连接异常: remoteAddress={}", ctx.channel().remoteAddress(), cause);
        super.exceptionCaught(ctx, cause);
    }



}
